package cd.itcast.spark.kudu.kudu_curd

import org.apache.kudu.spark.kudu.KuduContext
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, SparkSession}

object upsertData {

  def updateData(spark: SparkSession, kuduContext: KuduContext, tableName: String): Unit = {
    // a. 模拟产生数据
    val dfdata: DataFrame = spark.createDataFrame(
      Seq(
        (1001, "zhangsa风", 24, "男"),
        (1006, "tianqi", 33, "男")
      )
    ).toDF("id", "name", "age", "gender")
    // 数据存在做update；不存在报错
    kuduContext.upsertRows(dfdata,tableName)

  }

  def main(args: Array[String]): Unit = {
    //创建sparkconf实例
    val sparkConf: SparkConf = new SparkConf()
      .setAppName(this.getClass
        .getSimpleName
        .stripSuffix("$"))
      .setMaster("local[4]")
    //通过sparkconf实例创建sparksession实例对象
    val sparkSession: SparkSession = SparkSession
      .builder()
      .config(sparkConf)
      .getOrCreate()
    val kuduMaster: String="node01:7051,node02:7051,node03:7051"
    val kuduContext = new KuduContext(kuduMaster,sparkSession.sparkContext)
    val tableName:String="kudu_itcast"

    // 更新数据
    updateData(sparkSession, kuduContext, tableName)
    sparkSession.close()
  }


}
